package com.watson.kafka.lg.controller;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @author watson
 * @date 2021/7/29 9:31
 * @Description: TODO
 **/
@Component
public class KafkaConsumer {
    // 消费监听----过滤消息
    @KafkaListener(topics = {"topic1"}, containerFactory = "filterContainerFactory")
    public void onMessage1(ConsumerRecord<?, ?> record) {
        // 消费的哪个topic、partition的消息,打印出消息内容
        System.out.println("简单消费：" + record.topic() + "-" + record.partition() + "-" + record.value());
    }


    /**
     * @return void
     * @Title 指定topic、partition、offset消费
     * @Description 同时监听topic1和topic2，监听topic1的0号分区、topic2的 "0号和1号" 分区，指向1号分区的offset初始值为8
     * 属性解释：
     * id：消费者ID；
     * groupId：消费组ID；
     * topics：监听的topic，可监听多个；
     * topicPartitions：可配置更加详细的监听信息，可指定topic、parition、offset监听。
     * 上面onMessage2监听的含义：监听topic1的0号分区，同时监听topic2的0号分区和topic2的1号分区里面offset从8开始的消息。
     * 注意：topics和topicPartitions不能同时使用；
     **/
    @KafkaListener(id = "consumer1", groupId = "felix-group", errorHandler = "consumerAwareErrorHandler",
            topicPartitions = {
                    @TopicPartition(topic = "topic1", partitions = {"0"}),
                    @TopicPartition(topic = "topic2", partitions = "0",
                            partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "8"))})
    public void onMessage2(ConsumerRecord<?, ?> record) {
        System.out.println("topic:" + record.topic() + "|partition:" + record.partition() + "|offset:" + record.offset() + "|value:" + record.value());
    }

    /**
     * 批量消费
     * <p>
     * 设置application.prpertise开启批量消费即可，
     * # 设置批量消费
     * spring.kafka.listener.type=batch
     * # 批量消费每次最多消费多少条消息
     * spring.kafka.consumer.max-poll-records=50
     * 开启批量消费，接收消息时用List来接收，监听代码如下，
     *
     * @param records
     */
//    @KafkaListener(id = "consumer2", groupId = "felix-group", topics = "topic1", errorHandler = "consumerAwareErrorHandler")
    public void onMessage3(List<ConsumerRecord<?, ?>> records) {
        System.out.println(">>>批量消费一次，records.size()=" + records.size());
        for (ConsumerRecord<?, ?> record : records) {
            System.out.println(record.value());
        }
    }


    /*消息转发
    在实际开发中，我们可能有这样的需求，应用A从TopicA获取到消息，经过处理后转发到TopicB，
    再由应用B监听处理消息，即一个应用处理完成后将该消息转发至其他应用，完成消息的转发。
    在SpringBoot集成Kafka实现消息的转发也很简单，只需要通过一个@SendTo注解，
    被注解方法的return值即转发的消息内容，如下，*/

    /**
     * @return void
     * @Title 消息转发
     * @Description 从topic1接收到的消息经过处理后转发到topic2
     **/
    @KafkaListener(topics = {"topic1"})
    @SendTo("topic2")
    public String onMessage7(ConsumerRecord<?, ?> record) {
        return record.value() + "-forward message";
    }
}
